package net.xuele.learn.zk;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by huanghao（HIKARI） on 2019/6/11 0011.
 */
public class Curator {
    public static void main(String[] args) throws Exception {
        // 创建会话
        // RetryPolicy重试策略，默认实现有四种：ExponentialBackoffRetry、RetryNTimes、RetryOneTime、RetryUntilElapsed
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 3);
        //CuratorZookeeperClient curatorZookeeperClient = new CuratorZookeeperClient("1",1000,1000,null,policy);
        CuratorFramework client = CuratorFrameworkFactory.newClient("", 100, 100, policy);
        // 调用start方法完成会话的创建
        client.start();

        // fluent风格
        createZK(policy,"");
    }

    private static void createZK(RetryPolicy policy, String path) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("")
                .sessionTimeoutMs(100)
                .retryPolicy(policy)
                .namespace("base")// 创建隔离命名空间的会话，表明该客户端后续的操作都是基于这个目录/base进行操作的
                .build();
        client.start();
        // 创建一个节点，初始化内容为空，默认为持久化节点
        String s = client.create().forPath(path);
        // 创建一个节点，附带初始化内容
        String s1 = client.create().forPath(path, "init".getBytes());
        // 创建一个临时节点，初始化内容为空
        client.create().withMode(CreateMode.EPHEMERAL).forPath(path);
        // 创建一个临时节点，并自动递归创建父节点，zk规定所有非叶子节点必须为持久节点
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
        int version = 7;
        /**********************删除*************************/
        client.delete().forPath(path);
        client.delete().deletingChildrenIfNeeded().forPath(path);
        // 指定版本删除数据
        client.delete().withVersion(version).forPath(path);
        // 删除一个节点，强制保证删除，guaranteed()接口是一个保障措施，
        // 只要客户端会话有效，那么程序就会在后台持续进行删除操作，直到节点删除成功
        client.delete().guaranteed().forPath(path);

        /**********************读取数据*************************/
        byte[] bytes = client.getData().forPath(path);
        // 获取数据，并获取它的stat信息
        Stat stat = new Stat();
        bytes = client.getData().storingStatIn(stat).forPath(path);

        /**********************更新数据*************************/
        stat = client.setData().forPath(path);
        // 强制指定版本进行更新,withVersion就是用来实现CAS的
        client.setData().withVersion(stat.getVersion()).forPath(path);

        /**********************异步接口*************************/
        creteZKAsync(path);

    }

    /**
     *  涉及到的接口为{@link org.apache.curator.framework.api.BackgroundCallback}
     *  <br>在zk中，所有的异步通知事件都是由EventThread这个线程来处理的——EventThread线程用于串行处理所有的事件通知。
     *  <br> EventThread的“串行处理机制”在绝大部分应用场景中都能够保证对事件处理的顺序性，但这个特性也有其弊端，
     *  <br>就是碰到复杂的处理单元，会影响其他事件的处理。
     * @param path
     */
    private static void creteZKAsync(String path) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("")
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        CountDownLatch latch = new CountDownLatch(2);
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        // 传入了线程池，使用此线程池来执行此逻辑
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground((curatorFramework, curatorEvent) -> {
            System.out.println("event[code:"+curatorEvent.getResultCode()+",type："+curatorEvent.getType()+"]");
            latch.countDown();
        },executorService).forPath(path,"init".getBytes());
        latch.await();
        executorService.shutdown();
    }

    /**
     * 事件监听<br/>
     * 原生的是通过注册watcher来进行事件监听，但是不太方便，需要反复的注册watcher。
     * Cache是这个框架中对事件的封装，分为两种类型：节点监听和子节点监听
     * @param path
     */
    private void eventListening(String path) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("")
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path,"init".getBytes());
        // NodeCache不仅可以监听数据节点的内容变更，也能监听指定节点是否存在。
        // 如果原本节点不存在，那么cache就会在节点被创建后触发NodeCacheListener，但是数据节点被删除，则无法触发NodeCacheListener
        NodeCache nodeCache = new NodeCache(client, path, false);
        // 如果设置为true，那么NodeCache启动时，就会立刻从zk上读取对应节点的数据内容，并保持在cache中
        nodeCache.start(true);
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("Node Data update,new data :"+ new String(nodeCache.getCurrentData().getData()));
            }
        });

        client.setData().forPath(path);

        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
        pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                switch (pathChildrenCacheEvent.getType()){
                    case CHILD_ADDED:
                        System.out.println("CHILD_ADDED:" + pathChildrenCacheEvent.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        System.out.println("CHILD_REMOVED:"+pathChildrenCacheEvent.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        System.out.println("CHILD_UPDATED:"+pathChildrenCacheEvent.getData().getPath());
                        break;
                    default:
                        break;
                }

            }
        });

        // CHILD_ADDED
        client.create().withMode(CreateMode.PERSISTENT).forPath(path,"init".getBytes());
        // CHILD_REMOVED
        client.delete().guaranteed().forPath(path);
    }

    /**
     *  master选举：对于一个复杂的任务，仅需要从集群中选取一台机器来执行。此类问题被统称为master选举。
     *   思路与分布式锁类似：多台机器向某个节点创建子节点，创建成功的则是“master”
     * @param path 选举的根结点
     */
    private void selectMaster(String path) {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("")
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();
        LeaderSelector selector = new LeaderSelector(client, path, new LeaderSelectorListenerAdapter() {
            // 获取master权利成功之后会回调该方法，执行完该方法就会释放master权利
            @Override
            public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
                System.out.println("成为master节点");
                Thread.sleep(10000);
                System.out.println("完成master操作，释放权限");
            }
        });
        selector.autoRequeue();
        selector.start();
    }

    /**
     *  分布式锁
     * @param path
     */
    private void Lock(String path) {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("")
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();
        InterProcessMutex mutex = new InterProcessMutex(client, path);
        final CountDownLatch latch = new CountDownLatch(1);
        for (int i = 0; i < 30; i++) {
            new Thread(() -> {
                try {
                    latch.await();
                    mutex.acquire();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                SimpleDateFormat format = new SimpleDateFormat("HH:mm:ss");
                String orderNo = format.format(new Date());
                System.out.println("订单号为："+orderNo);
                try {
                    mutex.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
        // 模拟多线程
        latch.countDown();
    }

    /**
     * 分布式计数器
     * @param path
     */
        private void Counter(String path) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()

                .connectString("")
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();
        DistributedAtomicInteger atomicInteger = new DistributedAtomicInteger(client, path, new RetryNTimes(3, 1000));
        AtomicValue<Integer> rc = atomicInteger.add(8);
        System.out.println("result:"+ rc.succeeded());
    }

    /**
     * 分布式Barrier
     * @param path
     */
    private void Barrier(String path) {
        CyclicBarrier barrier = new CyclicBarrier(3);

    }

}